package org.alvin.dispatchtcp.handler;

import com.google.common.collect.Maps;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.ReferenceCountUtil;
import org.alvin.dispatchtcp.bean.DispatchServer;
import org.alvin.dispatchtcp.bean.ServerNode;

import java.net.SocketAddress;
import java.util.concurrent.ConcurrentMap;

/**
 * 代理服務
 */

public class DispatchServerHandler extends ChannelHandlerAdapter implements ChannelInboundHandler, ChannelOutboundHandler {

    private final DispatchServer dispatchServer;

    private ConcurrentMap<String, Channel> dispatchServerMap = Maps.newConcurrentMap();

    public DispatchServerHandler(DispatchServer dispatchServer) {
        this.dispatchServer = dispatchServer;
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
//        ctx.fireChannelRegistered();
//        for (ServerNode serverNode : this.dispatchServer.getServers()) {
//            EventLoopGroup workerGroup = new NioEventLoopGroup();
//            Bootstrap targetBootstrap = new Bootstrap();
//            targetBootstrap.group(workerGroup)
//                    .channel(NioSocketChannel.class)
//                    .handler(new ChannelInitializer<SocketChannel>() {
//                        @Override
//                        protected void initChannel(SocketChannel socketChannel) throws Exception {
//                            // 处理来自服务端的响应信息
//                            socketChannel.pipeline().addLast(new DispatchClientHandler(dispatchServer ,dispatchServerMap ,));
//                        }
//                    });
//
//            ChannelFuture targetFuture = targetBootstrap.connect(serverNode.getIp(), serverNode.getPort()).sync();
//            String uniqueId = targetFuture.channel().remoteAddress().toString();
//            String key = serverNode.getIp() + "_" + serverNode.getPort() + "_" + uniqueId;
//            dispatchServerMap.put(key, targetFuture.channel());
//        }
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelUnregistered();
        for (ServerNode serverNode : this.dispatchServer.getServers()) {
            try {
                String key = serverNode.getIp() + "_" + serverNode.getPort() + "_" + ctx.channel().remoteAddress().toString();
                Channel channel = dispatchServerMap.get(key);
                if (channel != null) {
                    channel.disconnect();
                    channel.close();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelInactive();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ctx.fireChannelRead(msg);
        try {
            if (msg instanceof ByteBuf) {
                ByteBuf data = (ByteBuf) msg;
                byte[] databyte = new byte[data.readableBytes()];
                data.readBytes(databyte);
                for (ServerNode serverNode : this.dispatchServer.getServers()) {
                    try {
                        String key = serverNode.getIp() + "_" + serverNode.getPort() + "_" + ctx.channel().remoteAddress().toString();
                        Channel channel = dispatchServerMap.get(key);
                        channel.writeAndFlush(databyte);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                data.discardReadBytes(); //丢弃已读的字节
            }
        } finally {
            // 必须释放msg数据
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelReadComplete();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        ctx.fireUserEventTriggered(evt);
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelWritabilityChanged();
    }


    @Override
    public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception {
        ctx.bind(localAddress, promise);
    }

    @Override
    public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception {
        ctx.connect(remoteAddress, localAddress, promise);
    }

    @Override
    public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        ctx.disconnect(promise);
    }

    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        ctx.close(promise);
    }

    @Override
    public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        ctx.deregister(promise);
    }

    @Override
    public void read(ChannelHandlerContext ctx) throws Exception {
        ctx.read();
    }

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        ctx.write(msg, promise);
    }

    @Override
    public void flush(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

}
